package com.atguigu.table

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table, TableEnvironment}
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors.{Csv, FileSystem, OldCsv, Schema}

/**
 * @ClassName TableAPITest
 * @Description
 * @Author Mr Yang
 * @Date 2020/10/5 11:52
 * @Version 1.0
 */
object TableAPITest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //创建表执行环境
    val tableEnv = StreamTableEnvironment.create(env)

    //老板本planner的流式查询
  /*  val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useOldPlanner()  //老版本
      .inStreamingMode()  //流处理
      .build()
    //调用
    val oldStreamTableEnv = StreamTableEnvironment.create(env,settings)
    //老版本批处理环境
    val batchEnv: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    val batchTableEnv: BatchTableEnvironment = BatchTableEnvironment.create(batchEnv)

    //blink版本的流式查询
    val bsSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val bsTableEnv = StreamTableEnvironment.create(env, bsSettings)
    //blink版本的批处理
    val bbSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inBatchMode()
      .build()
    val bbTableEnv = TableEnvironment.create(bbSettings)*/

    //读取数据
    val filePath = "F:\\work\\FlinkTutorial\\src\\main\\resources\\sensor.txt"
    tableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
          .field("id", DataTypes.STRING())
          .field("timestamp", DataTypes.BIGINT())
          .field("temperature", DataTypes.DOUBLE())
      )
      .createTemporaryTable("inputTable")

    //测试输出简单方式，需要将table转换成流
    val inputTable: Table = tableEnv.from("inputTable")
    val tableStream = inputTable.toAppendStream[(String, Long, Double)]

    tableStream.print()
    env.execute("table api test job")
  }
}
